// Project name:JavaRedis
// Creator:muhong
// Date time:2022/12/8,10:14 下午
// Name:AbstractNioByteChannel

package org.gmh.channel.nio;

import org.gmh.channel.Channel;
import org.gmh.channel.ChannelFuture;
import org.gmh.channel.ChannelInputShutdownEvent;
import org.gmh.channel.handler.ChannelPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;

public abstract class AbstractNioByteChannel extends AbstractNioChannel {

    private final static Logger logger = LoggerFactory.getLogger(AbstractNioByteChannel.class);

    protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        super(parent, ch, SelectionKey.OP_READ);
    }

    protected abstract int doReadBytes(ByteBuffer byteBuffer) throws Exception;

    @Override
    protected void doWrite(Object msg) throws Exception {
        SocketChannel channel = (SocketChannel) javaChannel();
        if (msg instanceof ByteBuffer) {
            ByteBuffer byteBuffer = (ByteBuffer) msg;
            byteBuffer.flip();
            channel.write(byteBuffer);
            byteBuffer.compact();
        } else if (msg instanceof byte[]) {
            byte[] contentBytes = (byte[]) msg;
            ByteBuffer buffer = ByteBuffer.allocate(contentBytes.length);
            buffer.put(contentBytes);
            buffer.flip();
            channel.write(buffer);
            buffer.compact();
        }
    }

    @Override
    protected AbstractUnsafe newUnsafe() {
        return new NioByteUnsafe();
    }

    @Override
    protected void doClose() throws Exception {

    }

    protected abstract ChannelFuture shutdownInput();

    protected boolean isInputShutdown0() {
        return false;
    }

    private static boolean isAllowHalfClosure() {
        return false;
    }

    protected class NioByteUnsafe extends AbstractNioUnsafe {

        @Override
        public void read() {
            logger.info("start to read some data from client");
            final ChannelPipeline pipeline = pipeline();
            ByteBuffer buffer = ByteBuffer.allocate(100);
            boolean close = false;
            try {
                int size = doReadBytes(buffer);
                close = size < 0;
                if (close) {
                    logger.info("client receive msg size < 0, close client channel");
                    closeOnRead(pipeline);
                    return;
                }
                pipeline.fireChannelRead(buffer);
            } catch (Throwable cause) {
                handleReadException(pipeline, buffer, cause, close);
            } finally {
                if (close) {
                    removeReadOp();
                }
            }
        }

        private void closeOnRead(ChannelPipeline pipeline) {
            if (!isInputShutdown0()) {
                if (isAllowHalfClosure()) {
                    shutdownInput();
                    pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
                } else {
                    close(voidPromise());
                }
            } else {
                pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
            }
        }

        private void handleReadException(ChannelPipeline pipeline, ByteBuffer byteBuffer, Throwable cause,
                                         boolean close) {
            if (byteBuffer != null) {
                if (byteBuffer.remaining() > 0) {
                    pipeline.fireChannelRead(byteBuffer);
                } else {
                    byteBuffer.clear();
                }
            }
            pipeline.fireExceptionCaught(cause);
            if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
                closeOnRead(pipeline);
            }
        }
    }
}
